package com.esb.transport.channel;

import java.util.concurrent.atomic.AtomicLong;

import com.esb.transport.protocal.Future;
import com.esb.transport.protocal.FutureContext;
import com.esb.transport.protocal.Invocation;
import com.esb.transport.protocal.Request;
import com.esb.transport.protocal.Response;
import com.esb.transport.protocal.ServiceDispatcher;

public abstract class AbstractChannel implements Channel {

	private final static AtomicLong ID = new AtomicLong();

	private static long getId() {
		long i = ID.getAndIncrement();
		if (i > Long.MAX_VALUE)
			ID.set(0);
		return i;
	}

	private boolean isOpen = false;

	@Override
	public Future<Response> send(Invocation message) {
		message.setId(getId());
		Future<Response> future = FutureContext.sent(message);
		doSend(message);
		return future;
	}

	abstract protected void doSend(Invocation message);

	@Override
	public void receive(Invocation msg) {
		if (msg instanceof Request) {
			doSend(new Response(msg.getId(),
					ServiceDispatcher.dispatch((Request) msg)));
		} else if (msg instanceof Response)
			FutureContext.received((Response) msg);
	}

	@Override
	public boolean isOpen() {
		if (isOpen == true)
			return isOpen;
		synchronized (this) {
			if (isOpen == false)
				try {
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
		}
		return isOpen;
	}

	@Override
	public void opened() {
		isOpen = true;
		synchronized (this) {
			this.notifyAll();
		}
	}

	@Override
	public void closed() {
		isOpen = false;
	}
}
